package today.wxp.blog.spark.structed.streaming.util

import org.apache.kafka.common.serialization.Deserializer
import today.wxp.blog.spark.structed.streaming.pojo.PlayInfo

/**
  * 播放信息反序列化。
  * <p/>
  * Author   : wangxp
  * <p/>
  * DateTime : 2017/11/23 14:58
  */
class PlayInfoDeserializer extends Deserializer[PlayInfo]
{
  private var encoding = "UTF8"

  override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit =
  {
    val propertyName = if (isKey)
    {
      "key.deserializer.encoding"
    }
    else
    {
      "value.deserializer.encoding"
    }
    var encodingValue = configs.get(propertyName)
    if (encodingValue == null)
    {
      encodingValue = configs.get("deserializer.encoding")
    }
    if (encodingValue != null && encodingValue.isInstanceOf[String])
    {
      this.encoding = encodingValue.asInstanceOf[String]
    }
  }

  override def deserialize(topic: String, data: Array[Byte]): PlayInfo = DeSerializationUtil.deserialize[PlayInfo](new String(data, this.encoding))

  override def close(): Unit =
  {
  }
}
